package com.atcyj.gamll.realtime.common.base;

import com.atcyj.gamll.realtime.common.constant.Constant;
import com.atcyj.gamll.realtime.common.util.FlinkSourceUtil;
import com.atcyj.gamll.realtime.common.util.SQLUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: cyj
 * @date: 2024/8/16
 */
public abstract class BaseSqlApp {
    public abstract void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId);


    public void start(int port, int parallelism, String ckAndGroupId) {
        // 1. 环境准备
        // 1.1 设置操作 Hadoop 的用户名为 Hadoop 超级用户 atcyj????
        System.setProperty("HADOOP_USER_NAME", "atcyj");

        // 1.2 获取流处理环境，并指定本地测试时启动 WebUI 所绑定的端口
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 1.3 设置并行度
        env.setParallelism(parallelism);

        // 1.4 状态后端及检查点相关配置
        // 1.4.1 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//
//        // 1.4.2 开启 checkpoint
//        env.enableCheckpointing(5000);
//        // 1.4.3 设置 checkpoint 模式: 精准一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 1.4.4 checkpoint 存储
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flink_ck/gmall2024/stream/" + ckAndGroupId);
//        // 1.4.5 checkpoint 并发数
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        // 1.4.6 checkpoint 之间的最小间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
//        // 1.4.7 checkpoint  的超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
//        // 1.4.8 job 取消时 checkpoint 保留策略
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(RETAIN_ON_CANCELLATION);

        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 执行具体的处理逻辑
        handle(env, tableEnv, ckAndGroupId);

    }

    /**
     * 读取kakfa topic_db主题数据，创建topic_db表
     *
     * @param ckAndGroupId
     * @param tableEnv
     */
    public void createTopicDB(String ckAndGroupId, StreamTableEnvironment tableEnv) {
        tableEnv.executeSql(SQLUtil.getKafkaTopicDB(ckAndGroupId));
    }

    /**
     * 读取HBase中dim_base_dic，创建对应的表
     *
     * @param tableEnv
     */
    public void createBaseDic(TableEnvironment tableEnv) {
        tableEnv.executeSql(
//                "create table base_dic (" +
//                        " dic_code string," +  // 如果字段是原子类型,则表示这个是 rowKey, 字段随意, 字段类型随意
//                        " info row<dic_name string>, " +  // 字段名和 hbase 中的列族名保持一致. 类型必须是 row. 嵌套进去的就是列
//                        " primary key (dic_code) not enforced " + // 只能用 rowKey 做主键
//                        ") WITH (" +
//                        " 'connector' = 'hbase-2.2'," +
//                        " 'table-name' = 'gmall:dim_base_dic'," +
//                        " 'zookeeper.quorum' = 'hadoop102:2181,hadoop103:2181,hadoop104:2181', " +
//                        " 'lookup.cache' = 'PARTIAL', " +
//                        " 'lookup.async' = 'true', " +
//                        " 'lookup.partial-cache.max-rows' = '20', " +
//                        " 'lookup.partial-cache.expire-after-access' = '2 hour' " +
//                        ")");

                "CREATE TABLE base_dic (\n" +
                        " dic_code STRING,\n" +
                        " info ROW<dic_name STRING>,\n" +
                        " PRIMARY KEY (dic_code) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        " 'connector' = 'hbase-2.2',\n" +
                        " 'table-name' = 'gmall:dim_base_dic',\n" +
                        " 'zookeeper.quorum' = '" + Constant.HBASE_ZOOKEEPER_QUORUM + "'\n" +
                        ");");
    }


}
